package com.zhu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhu.bean.TrafficHomeDetailPageViewBean;
import com.zhu.utils.ClickHouseUtil;
import com.zhu.utils.DataFormatUtil;
import com.zhu.utils.ZhuKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 流量域页面浏览各窗口汇总表
 * 运用 Flink 状态编程，为每个 mid 维护首页和商品详情页末次访问日期。如果 page_id
 * 为 home，当状态中存储的日期为 null 或不是当日时，将 homeUvCt（首页独立访客数） 置
 * 为 1，并将状态中的日期更新为当日。否则置为 0，不做操作。商品详情页独立访客的统计
 * 同理。当 homeUvCt 和 detailUvCt 至少有一个不为 0 时，将统计结果和相关维度信息封
 * 装到定义的实体类中，发送到下游，否则舍弃数据。
 */
public class DWSTrafficPageViewWindowApp {

    public static void main(String[] args) throws Exception {

        //todo env
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setParallelism(4);

        //checkPoint
        /*
        streamExecutionEnvironment.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE); //精确一次
        //状态后端
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage(ClusterParametersConfig.HDFS_CHECKPOINT_FILE_DIR);  //检查点保存在hdfs
        System.setProperty("HADOOP_USER_NAME", "zhu");
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);  //TimeOut
        streamExecutionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(2);  //最大共存检查点
        streamExecutionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5 * 1000L));  //重启策略
         */

        //todo 读取 kafka 页面日志主题数据

        String topic = "dwd_traffic_page_log";
        String groupId = "dws_traffic_page_view_window";
        DataStreamSource<String> kafkaDStream = streamExecutionEnvironment.addSource(ZhuKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //todo 将每行数据转换为json 过首页和商品详情
        SingleOutputStreamOperator<JSONObject> jsonObjFilterDStream = kafkaDStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            //转化为json
            @Override
            public void flatMap(String value, Collector<JSONObject> collector) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String pageId = jsonObject.getJSONObject("page").getString("page_id");
                //过滤首页与商品详情数据
                if ("home".equals(pageId) || "good_detail".equals(pageId)) {
                    collector.collect(jsonObject);
                }
            }
        });

        //todo WaterMark
        SingleOutputStreamOperator<JSONObject> jsonObjWithWaterMarkDStream = jsonObjFilterDStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long l) {
                        return jsonObject.getLong("ts");
                    }
                }));

        //todo Mid分组
        KeyedStream<JSONObject, String> keyedStream =
                jsonObjWithWaterMarkDStream.keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"));

        //todo 状态编程过滤独立访客
        //flatMap process
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> uvStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>() {

            private ValueState<String> homeLastVisitDt;
            private ValueState<String> detailLastVisitDt;

            @Override
            public void open(Configuration parameters) throws Exception {


                super.open(parameters);

                StateTtlConfig stateTtlConfig = new StateTtlConfig.Builder(org.apache.flink.api.common.time.Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build();
                ValueStateDescriptor<String> detailLastVisitDescriptor =
                        new ValueStateDescriptor<>("detail-state", String.class);
                ValueStateDescriptor<String> homeLastVisitDescriptor =
                        new ValueStateDescriptor<String>("home-state", String.class);

                detailLastVisitDescriptor.enableTimeToLive(stateTtlConfig);
                homeLastVisitDescriptor.enableTimeToLive(stateTtlConfig);

                homeLastVisitDt = getRuntimeContext().getState(homeLastVisitDescriptor);
                detailLastVisitDt = getRuntimeContext().getState(detailLastVisitDescriptor);

                //设置ttl


            }


            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject,
                    TrafficHomeDetailPageViewBean>.Context context, Collector<TrafficHomeDetailPageViewBean> collector) throws Exception {


                String homeLastDt = homeLastVisitDt.value();
                String detailLastDt = detailLastVisitDt.value();


                JSONObject page = jsonObject.getJSONObject("page");
                String page_id = page.getString("page_id");
                Long ts = jsonObject.getLong("ts");
                String visitDt = DataFormatUtil.toDate(ts);

                long homeUVCt = 0L;
                long detailUVCt = 0L;

                if("home".equals(jsonObject.getJSONObject("page").getString("page_id"))) {
                    if (homeLastDt == null || !homeLastDt.equals(visitDt)) {
                        homeUVCt = 1L;
                        homeLastVisitDt.update(visitDt);
                    }
                }else {
                    if (detailLastDt == null || !detailLastDt.equals(visitDt)) {
                        detailUVCt = 1L;
                        detailLastVisitDt.update(visitDt);

                    }
                }
                if (homeUVCt != 0 || detailUVCt != 0) {
                    collector.collect(new TrafficHomeDetailPageViewBean(
                            "", "", homeUVCt, detailUVCt, 0L
                    ));
                }
            }
        });

        //todo 开创聚合
        AllWindowedStream<TrafficHomeDetailPageViewBean, TimeWindow> allWindowDStream = uvStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> resultDStream
                = allWindowDStream.reduce(
                        new ReduceFunction<TrafficHomeDetailPageViewBean>() {
                    @Override
                    public TrafficHomeDetailPageViewBean reduce(TrafficHomeDetailPageViewBean value1, TrafficHomeDetailPageViewBean value2) throws Exception {

                        value1.setGoodDetailUvCt(value1.getGoodDetailUvCt() + value2.getGoodDetailUvCt());
                        value1.setHomeUvCt(value1.getHomeUvCt() + value2.getHomeUvCt());
                        return value1;
                        }},
                        new AllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<TrafficHomeDetailPageViewBean> values, Collector<TrafficHomeDetailPageViewBean> collector) throws Exception {

                        String windowStartTime = DataFormatUtil.toYmdHms(timeWindow.getStart());
                        String windowEndTime = DataFormatUtil.toYmdHms(timeWindow.getEnd());

                        TrafficHomeDetailPageViewBean viewBean = values.iterator().next();
                        viewBean.setTs(System.currentTimeMillis());
                        viewBean.setStt(windowStartTime);
                        viewBean.setEdt(windowEndTime);
                        collector.collect(viewBean);
                    }
                });

        //todo 将数据写出到ClickHouse
        resultDStream.print(">>>");
        resultDStream.addSink(ClickHouseUtil.getClickHouseSinkFunction("" +
                "insert into dws_traffic_page_view_window values(?,?,?,?,?)"));

        //todo execute
        streamExecutionEnvironment.execute("DWSTrafficPageViewWindowApp");
    }
}
